package com.firesrc.mail.thread;

import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import jodd.db.DbThreadSession;
import jodd.db.oom.DbOomQuery;
import jodd.util.collection.SimpleQueue;

import com.firesrc.bean.Record;
import com.firesrc.db.DBManager;
import com.firesrc.mail.send.MailSendManagerFactory;
import com.firesrc.mail.send.MailSender;

public class SendMailCompleteExecutor {
	private static ExecutorService executor = Executors.newFixedThreadPool(5);

	public static void main(String[] args) throws Throwable {
		SendMailCompleteExecutor completeExecutor = new SendMailCompleteExecutor();
		completeExecutor.execute();
		executor.shutdown();
	}

	public void execute() throws Throwable {
		MailSendManagerFactory managerFactory = new MailSendManagerFactory();
		final List<MailSender> mailSenderList = managerFactory.getList();
		CompletionService<MailSender> completionService = new ExecutorCompletionService<MailSender>(executor);
		for (final MailSender mailSender : mailSenderList)
			completionService.submit(new CallTask(mailSender));

		System.out.println("SendMailThread.run() 开始发邮件:" + new Date());
		new DbThreadSession(DBManager.getPublicConnectionPool());
		DbOomQuery q = new DbOomQuery("select top 20 * from record where mailstate is null and mail is not null");
		List<Record> recordList = q.list(Record.class);
		SimpleQueue<Record> recordQueue = new SimpleQueue<Record>();

		for (Record record : recordList) {
			recordQueue.put(record);
		}

		SimpleQueue<MailSender> successMailSenderQueue = new SimpleQueue<MailSender>();
		try {
			for (int t = 0, n = mailSenderList.size(); t < n; t++) {
				Future<MailSender> f = completionService.take();
				MailSender mailSender = f.get();
				if (mailSender != null) {
					mailSender.batchSendMail(recordQueue);
					successMailSenderQueue.put(mailSender);
				}
			}
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		} catch (ExecutionException e) {
			throw new Throwable(e.getCause());
		}

		while (!recordQueue.isEmpty()) {
			MailSender superMailSender = successMailSenderQueue.get();
			superMailSender.batchSendMail(recordQueue);
			successMailSenderQueue.put(superMailSender);
		}

		for (final MailSender mailSender : mailSenderList) {
			mailSender.closeSession();
		}
		System.out.println("SendMailCompleteExecutor.execute()  邮件发送完成");
	}

	private final class CallTask implements Callable<MailSender> {
		private final MailSender mailSender;

		private CallTask(MailSender mailSender) {
			this.mailSender = mailSender;
		}

		public MailSender call() {
			return mailSender.openSession();
		}
	}

}
